package dataStream;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/10/12
 */
public class AssignWatermark {
    public static void main(String[] args) {
        List<Person> data = new ArrayList<>();
        data.add( new Person("man", 20, 12345L));
        data.add( new Person("man", 18, 123712L));
        data.add( new Person("man", 16, 12998790L));
        data.add( new Person("woman", 20, 1827961387L));
        data.add( new Person("woman", 18, 12983789L));
        data.add( new Person("woman", 16, 1287637812L));
        data.add( new Person("woman", 15,897126L));

        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        // 官方提供 watermark 生成策略，在 watermark 中指定事件时间
        DataStream<Person> dataStream = env.fromCollection(data);
        dataStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Person>forBoundedOutOfOrderness(Duration.ofMillis(20L))
                        .withTimestampAssigner((event, timestamp)->event.getTimestamp())
        );

        // 自定义周期性 watermark 生成策略
        dataStream.assignTimestampsAndWatermarks(new WatermarkStrategy<Person>() {
            @Override
            public WatermarkGenerator<Person> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Person>() {

                    private final long maxOutOfOrderness = 3500; // 3.5 seconds

                    private long currentMaxTimestamp;

                    // 基于事件生成 watermark
                    @Override
                    public void onEvent(Person event, long eventTimestamp, WatermarkOutput output) {
                        currentMaxTimestamp = Math.max(currentMaxTimestamp, eventTimestamp);
                    }
                    // 基于周期生成 watermark
                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness - 1));
                    }
                };
            }
        });

        // 自定义事件 watermark 生成策略
        dataStream.assignTimestampsAndWatermarks(new WatermarkStrategy<Person>() {
            @Override
            public WatermarkGenerator<Person> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Person>() {

                    // 基于事件生成 watermark
                    @Override
                    public void onEvent(Person event, long eventTimestamp, WatermarkOutput output) {
                        if (event.getAge() > 10) {
                            output.emitWatermark(new Watermark(event.getTimestamp()));
                        }
                    }
                    // 基于周期生成 watermark
                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {

                    }
                };
            }
        });
    }
}
